package spark_core.operate_transform.keyvalue;

import java.util.Arrays;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;


/**
 * @author shihb
 * @date 2020/1/8 11:43
 * 根据key,进行聚合(分区和分区间可以不同操作)
 */
public class AggregateByKeyDemo {

  public static void main(String[] args) {
    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("mark rdd");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext sc = new JavaSparkContext(sparkConf);

    //key-value的方法要pairRdd才能调用,所以创建JavaPairRDD
    JavaPairRDD<String,Integer> inputPairRDD = sc.parallelizePairs(Arrays
        .asList(new Tuple2<>("a",1), new Tuple2("b", 2), new Tuple2("a",3),
            new Tuple2("b", 4),new Tuple2("c", 5),new Tuple2("a", 6)),2);
    inputPairRDD.glom().collect().forEach(System.out::println);

//  [(a,1), (b,2), (a,3)]
//  [(b,4), (c,5), (a,6)]

    //需求同一个分区求最大值,分区间相加
    //参数(zeroValue初始值,seqOp同分区聚合函数，combOp分区间聚合函数)
    //可选参数:分区设置

    JavaPairRDD<String, Integer> aggRdd = inputPairRDD
        .aggregateByKey(0, (o1, o2) -> Math.max(o1, o2), (o1, o2) -> o1 + o2);

    aggRdd.glom().collect().forEach(System.out::println);
    //结果
    //[(b,6)]
    //[(a,9), (c,5)]]

    //停止
    sc.stop();

  }
}
